package com.wuchao.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.*;
import java.lang.reflect.Array;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;

/**
 * 作者：吴超
 * 邮箱：wuchao_19931120@163.com
 * 时间：2021-04-13
 * 描述：
 * 客户端代码常用套路：
 * 1.获取一个客户端对象
 * 2.用客户端对象做各种操作
 * 3.关闭客户端
 * <p>
 * 经典案例：HDFS和Zookeeper
 */

public class HdfsClient {

    private URI uri;
    private Configuration conf;
    private String user;
    private FileSystem fs;


    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {

        uri = new URI("hdfs://hadoop102:9820");
        conf = new Configuration();
        user = "wuchao";

        conf.set("dfs.replication", "2");

        //1.获取一个客户端对象

        System.out.println();


        fs = FileSystem.get(uri, conf, user);

    }

    @After
    public void close() throws IOException {
        //3.关闭客户端
        fs.close();
    }


    //2.用客户端对象做各种操作

    /**
     * 创建文件
     */
    @Test
    public void testHdfs() throws IOException, InterruptedException, URISyntaxException {

        boolean b = fs.mkdirs(new Path("/java2"));

    }


    /**
     * 文件上传
     * <p>
     * hadoop修改配置文件：
     * 1.xxx-default.xml
     * 2.xxx-site.xml
     * 3.客户端修改,项目的路径下写 xxx-site.xml
     * 4.通过conf设置
     * <p>
     * 优先级：1 < 2 < 3 < 4
     */
    @Test
    public void put() throws IOException {

        //fs.copyFromLocalFile(false,false,new Path("E:\\hadoop_java\\hello.txt"),new Path("/java"));
        fs.copyFromLocalFile(false, false, new Path("E:\\hadoop_java\\hello.txt"), new Path("/input"));

    }

    /**
     * 文件下载
     */
    @Test
    public void get() throws IOException {

        fs.copyToLocalFile(false, new Path("/input/hello.txt"), new Path("E:\\"), true);
    }

    /**
     * 文件/文件夹删除
     */
    @Test
    public void rm() throws IOException {
        //删除文件
        //fs.delete(new Path("/input/hello.txt"),false);

        //删除文件夹，后面参数为是不是递归，空目录就不需要递归
        fs.delete(new Path("/java2"), true);
    }

    /**
     * 文件更名和移动
     */

    @Test
    public void mv() throws IOException {
        //支持文件和文件加的更名和移动
        //fs.rename(new Path("/java/hello.txt"),new Path("/input/hello1.txt"));
        fs.rename(new Path("/input/hello1.txt"), new Path("/java/hello.txt"));
    }

    /**
     * 文件详细信息查看
     */
    @Test
    public void ls() throws IOException {

        RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(new Path("/input"), true);
        while (iterator.hasNext()) {
            LocatedFileStatus fileStatus = iterator.next();
            System.out.println("===========" + fileStatus.getPath() + "===========");
            System.out.println(fileStatus.getPermission());
            System.out.println(fileStatus.getOwner());
            System.out.println(fileStatus.getGroup());
            System.out.println(fileStatus.getLen());
            System.out.println(fileStatus.getModificationTime());
            System.out.println(fileStatus.getReplication());
            System.out.println(fileStatus.getBlockSize());
            System.out.println(fileStatus.getPath().getName());

            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            //System.out.println(blockLocations);
            System.out.println(Arrays.toString(blockLocations));

        }


    }

    /**
     * 文件和文件夹判断
     */

    @Test
    public void isFileOrDir() throws IOException {

        FileStatus[] fileStatuses = fs.listStatus(new Path("/"));
        for (FileStatus fileStatus : fileStatuses) {
            boolean file = fileStatus.isFile();
            if (file) {
                System.out.println("文件：" + fileStatus.getPath());

            } else {
                System.out.println("目录：" + fileStatus.getPath());
            }
        }
    }

    /**
     * 递归判断传入路径下的文件和目录
     */
    public void isAll(String path, FileSystem fileSystem) throws IOException {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path(path));
        for (FileStatus fileStatus : fileStatuses) {
            boolean file = fileStatus.isFile();
            if (file) {
                System.out.println("文件：" + fileStatus.getPath());
            } else {
                System.out.println("目录：" + fileStatus.getPath());
                isAll(fileStatus.getPath().toString(), fileSystem);
            }
        }
    }

    @Test
    public void testIsAll() throws IOException {
        isAll("/", fs);
    }


    /**
     * 基于IO流的上传
     */

    @Test
    public void putByIO() throws IOException {

        //1.获取本地文件输入流
        FileInputStream fis = new FileInputStream(new File("E:\\hadoop_java\\hello.txt"));

        //2.获取hdfs文件输出流
        FSDataOutputStream hdfsfos = fs.create(new Path("/input/hello.txt"));

        //3.流的对拷
        IOUtils.copyBytes(fis, hdfsfos, conf);

        //4.流的关闭
        IOUtils.closeStream(hdfsfos);
        IOUtils.closeStream(fis);
    }

    /**
     * 文件IO流的下载
     */

    @Test
    public void getByIO() throws IOException {

        FSDataInputStream hdfsfis = fs.open(new Path("/input/hello.txt"));

        FileOutputStream fos = new FileOutputStream(new File("E:\\hadoop_java\\hello1.txt"));

        IOUtils.copyBytes(hdfsfis, fos, conf);

        IOUtils.closeStream(fos);
        IOUtils.closeStream(hdfsfis);
    }

}

















